package com.tarena.rocketmq.study;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.nio.charset.StandardCharsets;
import java.sql.SQLOutput;
import java.util.List;

/**
 * 监听一个主题
 * 拿到消息进行消费
 */
public class MyConsumer {
    @Test
    public void consume() throws Exception{
    //1. 创建一个消费者对象
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer();
        //连接nameserver
        consumer.setNamesrvAddr("localhost:9876");
        //消费者分组
        consumer.setConsumerGroup("test-consumer01");
        //设置主题 监听这个主题 同时定义过滤tag表达式 *表示所有tag都满足
        consumer.subscribe("test-01-topic","北京");
    //2. 创建监听器 实现消费逻辑
        MessageListenerConcurrently messageListener=
                new MessageListenerConcurrently() {
                    /**
                     * 每拿到一条消息,方法就调用一次
                     * @param msgs 接收到的消息
                     * @param consumeConcurrentlyContext 消费者信息
                     * @return CONSUME_SUCCESS 表示消费成功 RECONSUME_LATER 表示过一段时间再重新消费
                     */
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                        //解析消息
                        MessageExt messageExt = msgs.get(0);
                        byte[] body = messageExt.getBody();
                        String msg=new String(body, StandardCharsets.UTF_8);
                        System.out.println("消费者消费:"+msg);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                };
        //给消费者装配这个监听器
        consumer.setMessageListener(messageListener);
    //3. 建立连接开启消费者
        consumer.start();
        while(true);
    }
    @Test
    public void consume01() throws Exception{
        //1. 创建一个消费者对象
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer();
        //连接nameserver
        consumer.setNamesrvAddr("localhost:9876");
        //消费者分组
        consumer.setConsumerGroup("test-consumer02");
        //设置主题 监听这个主题
        consumer.subscribe("test-01-topic","*");
        //2. 创建监听器 实现消费逻辑
        MessageListenerConcurrently messageListener=
                new MessageListenerConcurrently() {
                    /**
                     * 每拿到一条消息,方法就调用一次
                     * @param msgs 接收到的消息
                     * @param consumeConcurrentlyContext 消费者信息
                     * @return CONSUME_SUCCESS 表示消费成功 RECONSUME_LATER 表示过一段时间再重新消费
                     */
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                        //解析消息
                        MessageExt messageExt = msgs.get(0);
                        byte[] body = messageExt.getBody();
                        String msg=new String(body, StandardCharsets.UTF_8);
                        System.out.println("消费者消费:"+msg);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                };
        //给消费者装配这个监听器
        consumer.setMessageListener(messageListener);
        //3. 建立连接开启消费者
        consumer.start();
        while(true);
    }
}
